package com.stars.easyms.schedule.mq.activemq;

import com.stars.easyms.schedule.exception.DistributedScheduleMQException;
import com.stars.easyms.schedule.mq.MQMessageSender;
import org.springframework.jms.connection.JmsResourceHolder;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Session;

/**
 * ActiveMQ消息发送者
 *
 * @author guoguifang
 */
public class ActiveMQMessageSender implements MQMessageSender {

    private ConnectionFactory connectionFactory;

    private JmsMessagingTemplate jmsMessagingTemplate;

    @Override
    public boolean convertAndSend(String destination, Object message) throws DistributedScheduleMQException {
        try {
            registerMQSenderSession();
            jmsMessagingTemplate.convertAndSend(destination, message);
            return true;
        } catch (DistributedScheduleMQException e) {
            throw e;
        } catch (Exception e) {
            throw new DistributedScheduleMQException("Use activeMQ send message to destination [" + destination + "] failure!", e);
        }
    }

    private void registerMQSenderSession() throws DistributedScheduleMQException {
        // 注册MQ发送事务资源
        try {
            if (TransactionSynchronizationManager.getResource(connectionFactory) == null) {
                Connection connection = connectionFactory.createConnection();
                connection.start();
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                TransactionSynchronizationManager.bindResource(connectionFactory, new JmsResourceHolder(connectionFactory, connection, session));
            }
        } catch (Exception e) {
            throw new DistributedScheduleMQException("Registered activeMQ send transaction resource failed!", e);
        }
    }

    public ActiveMQMessageSender(String brokerUrl, String user, String password) {
        this.connectionFactory = BasicActiveMQConnectionFactory.getActiveMQConnectionFactory(brokerUrl, user, password);
        this.jmsMessagingTemplate = new JmsMessagingTemplate(connectionFactory);
    }

}
